package UDF

import Source.SensorReading
import org.apache.flink.api.common.functions.FilterFunction
import org.apache.flink.streaming.api.scala._

class MyFilterFunction extends FilterFunction[SensorReading] {
  override def filter(t: SensorReading): Boolean =
    t.id.startsWith("sensor_1")
}

object MyFilterTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val inputPath = "src/main/resources/SensorReading"
    val inputStream = env.readTextFile(inputPath)

    //转换成样例类类型
    val dataStream = inputStream.map(
      data => {
        val arr = data.split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      }
    )
    //    val result = dataStream.filter(new MyFilterFunction)
    val result = dataStream.filter(new FilterFunction[SensorReading] {
      override def filter(t: SensorReading) = t.id.startsWith("sensor_1")
    })
    result.print()
    env.execute()
  }
}